package com.shujia.util;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.sql.Array;
import java.util.ArrayList;
import java.util.List;

public class HBaseTool {

    public static final Connection HBASE_CONN;
    public static final Admin HBASE_ADMIN;

    static  {
        Connection conn = null;
        Admin admin = null;
        try {
            //使用HBase依赖种的类来创建与hbase的连接对象
            //HBaseConfiguration Deprecated
            //0.90.0版本后已经被弃用. 新版本种使用create()方法替代
//        HBaseConfiguration conf = new HBaseConfiguration();
            Configuration conf = HBaseConfiguration.create();
            //设置zookeeper的集群
            conf.set("hbase.zookeeper.quorum", "master,node1,node2");

            //通过配置文件，获取与hbase的连接对象
            conn = ConnectionFactory.createConnection(conf);
            //获取数据库操作对象
            admin = conn.getAdmin();
            System.out.println("---------------------------------------");
        } catch (Exception e) {
            e.printStackTrace();
        }
        HBASE_CONN = conn;
        HBASE_ADMIN = admin;
    }

    private HBaseTool(){}

    public static void createOneTable(String tableName, String... cfs){
        try {
            //将表名封装成TableName对象
            //TableName valueOf(String name)
            TableName name = TableName.valueOf(tableName);
            TableDescriptorBuilder table = TableDescriptorBuilder.newBuilder(name);
            ArrayList<ColumnFamilyDescriptor> cfList = new ArrayList<>();
            for (String cf : cfs) {
                cfList.add(ColumnFamilyDescriptorBuilder.of(cf));

            }
            table.setColumnFamilies(cfList);
            HBASE_ADMIN.createTable(table.build());
            System.out.println("成功创建：" + name);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static Table getOneTable(String tableName) throws Exception{
        TableName name = TableName.valueOf(tableName);

        return HBASE_CONN.getTable(name);
    }

    public static void printOneData(Result result){
        try {
            //获取列数据的方式2：List<Cell> listCells() //不知道列名的情况下遍历
            //获取所有列的单元格类型，遍历list集合获取每一个单元格列
            List<Cell> cells = result.listCells();
            String rk = Bytes.toString(result.getRow());
            StringBuilder sb = new StringBuilder();
            sb.append("rowKey：").append(rk).append(", ");
            for (Cell cell : cells) {
                String familyName = Bytes.toString(CellUtil.cloneFamily(cell));
                String colName = Bytes.toString(CellUtil.cloneQualifier(cell));
                String colValue = Bytes.toString(CellUtil.cloneValue(cell));
                sb.append(familyName).append("_").append(colName).append("：").append(colValue).append(", ");
            }
            String info = sb.toString();
            info = info.substring(0,info.length()-2);
            System.out.println(info);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void printMoreData(ResultScanner resultScanner){
        try {
            //调用iterator方法获取包含所有行数据的迭代器
            for (Result result : resultScanner) {
                printOneData(result);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }



}
